package queue

import (
	"context"
	"fmt"
	"time"

	"github.com/streadway/amqp"
	"golang-manage-admin/providers"
	"golang-manage-admin/queue/business"
)

func NewQueue(region string) *Queue {
	return &Queue{
		mqs:    make(map[string]*RabbitMq),
		infos:  make(map[string]*QueueInfo),
		Region: region,
	}
}

// 队列信息
type QueueInfo struct {
	IsDeadLetter   bool
	ExchangeName   string
	ExchangeType   string
	RoutingKey     string
	QueueName      string
	Qos            int
	Args           amqp.Table
	businessIQueue business.IQueue
}

type Queue struct {
	mqs    map[string]*RabbitMq
	infos  map[string]*QueueInfo
	Region string
}

/**
 * 队列绑定
 */
func (q *Queue) Bind(ctx context.Context) {
	for cKey, _ := range q.infos {
		//key := cKey
		//queue := cQueue
		// 追加关闭时的重新启动
		q.queueBind(ctx, cKey, 0)
	}
}

/**
 * 队列绑定
 */
func (q *Queue) queueBind(ctx context.Context, key string, retryCount int) {
	queue := q.infos[key]

	q.mqs[key] = NewRabbitMq(
		q.infos[key].ExchangeName,
		q.infos[key].ExchangeType,
		q.infos[key].RoutingKey,
		q.infos[key].QueueName,
		q.infos[key].Qos,
		q.Region,
		q.infos[key].Args,
		q.infos[key].IsDeadLetter,
	)

	// @TODO 还需要清理操作
	// 日志设置
	q.infos[key].businessIQueue.SetLogger(
		providers.Logger,
	)

	// mq 建立后初始化
	q.infos[key].businessIQueue.Init()
	// 新建对应的线程处理
	go func(key string) {
		providers.Logger.Infof("(%s)开始绑定", key)
		msgChan := q.mqs[key].Bind()
		for {
			select {
			case msg, ok := <-msgChan:
				//providers.Logger.Debugf("消息进入")
				if ok {
					go q.infos[key].businessIQueue.Delivery(ctx, msg)
				} else {
					providers.Logger.Infof("消息队列%s channel 已经关闭", key)
					goto CHANNELEXIT
				}
				break
			case <-ctx.Done():
				providers.Logger.Infof("队列关闭:%s", ctx.Err().Error())
				goto CHANNELEXIT
			}
		}

	CHANNELEXIT:
		// 队列断掉之后自动重新启动
		retryCount = retryCount + 1

		sleepSecond := time.Duration(retryCount * 2)
		providers.Logger.Infof("(%s)队列结束, %d 秒之后重新启动", key, sleepSecond)
		time.Sleep(time.Second * sleepSecond)

		// 关闭 mq
		q.infos[key].businessIQueue.Close()
		q.mqs[key].Close()
		q.mqs[key] = nil

		// 重新启动
		q.queueBind(ctx, key, retryCount)
	}(key)

	queue.businessIQueue.GetLogger().Infof("(%s)队列启动成功", key)
}

/*
 * 设置队列
 */
func (q *Queue) SetQueue(businessQueue business.IQueue) {
	configKey := "queue." + businessQueue.GetConfigKey()
	// 检测配置参数是否规范
	queueConfig := q.checkConfig(configKey)

	var args amqp.Table
	// 是否绑定死信队列
	if deadLetter, exist := queueConfig["dead_letter"]; exist && deadLetter.(bool) {
		switch queueConfig["dead_letter_type"] {
		case "QueueTTL": // Queue TTL 模式
			args = amqp.Table{
				"x-dead-letter-exchange":    queueConfig["dead-letter-exchange"].(string),
				"x-dead-letter-routing-key": queueConfig["dead-letter-routing-key"].(string),
				"x-expires":                 queueConfig["expires"].(int64),
			}
		case "MessageTTL": // Message TTL 模式： Per-Queue Message TTL
			args = amqp.Table{
				"x-dead-letter-exchange":    queueConfig["dead-letter-exchange"].(string),
				"x-dead-letter-routing-key": queueConfig["dead-letter-routing-key"].(string),
				"x-message-ttl":             queueConfig["expires"].(int64),
			}
		case "MessageTTL-Single": // Message TTL 模式： Per-Message TTL
			args = amqp.Table{
				"x-dead-letter-exchange":    queueConfig["dead-letter-exchange"].(string),
				"x-dead-letter-routing-key": queueConfig["dead-letter-routing-key"].(string),
			}
		}
	}

	queueInfo := QueueInfo{
		IsDeadLetter:   queueConfig["dead_letter"].(bool),
		ExchangeName:   queueConfig["exchange_name"].(string),
		ExchangeType:   queueConfig["exchange_type"].(string),
		QueueName:      queueConfig["queue_name"].(string),
		RoutingKey:     queueConfig["routing_key"].(string),
		Qos:            int(queueConfig["qos"].(int64)),
		Args:           args,
		businessIQueue: businessQueue,
	}
	switch queueConfig["exchange_type"] {
	case "direct":
		queueInfo.ExchangeType = amqp.ExchangeDirect
	case "fanout":
		queueInfo.ExchangeType = amqp.ExchangeFanout
	case "topic":
		queueInfo.ExchangeType = amqp.ExchangeTopic
	case "headers":
		queueInfo.ExchangeType = amqp.ExchangeHeaders
	}

	key := fmt.Sprintf(
		"%s:%s:%s:%s:%s",
		queueInfo.ExchangeName,
		queueInfo.ExchangeType,
		queueInfo.RoutingKey,
		queueInfo.QueueName,
		q.Region,
	)

	q.infos[key] = &queueInfo
}

/**
 *  检测配置参数是否有误
 */
func (q *Queue) checkConfig(configKey string) map[string]interface{} {
	// 获取队列配置
	queueInfo := providers.Config.GetStringMap(configKey)
	providers.Logger.Debugf("key:%s;map:%+v", configKey, queueInfo)

	if _, exist := queueInfo["exchange_name"]; !exist {
		panic(fmt.Sprintf("%s;缺少exchange_name", configKey))
	}
	if exchangeType, exist := queueInfo["exchange_type"]; !exist || (exchangeType.(string) != "direct" &&
		exchangeType.(string) != "fanout" &&
		exchangeType.(string) != "topic" &&
		exchangeType.(string) != "x-delayed-message" &&
		exchangeType.(string) != "headers") {
		panic(fmt.Sprintf("%s;缺少exchange_type", configKey))
	}
	if _, exist := queueInfo["routing_key"]; !exist {
		panic(fmt.Sprintf("%s;缺少routing_key", configKey))
	}
	if _, exist := queueInfo["queue_name"]; !exist {
		panic(fmt.Sprintf("%s;缺少queue_name", configKey))
	}

	// 死信队列配置
	if deadLetter, exist := queueInfo["dead_letter"]; exist && deadLetter.(bool) {
		if deadLetterType, exist := queueInfo["dead_letter_type"]; !exist || (deadLetterType.(string) != "QueueTTL" &&
			deadLetterType.(string) != "MessageTTL" &&
			deadLetterType.(string) != "MessageTTL-Single") {

			panic(fmt.Sprintf("%s;缺少dead_letter_type", configKey))
		}
	}
	if deadLetterType, exist := queueInfo["dead_letter_type"]; exist && (deadLetterType.(string) == "QueueTTL" ||
		deadLetterType.(string) == "MessageTTL") {
		// Queue TTL 模式 和 Per-Queue Message TTL 模式 需要queue中配置过期时间
		if expires, exist := queueInfo["expires"]; !exist || expires.(int64) <= 0 {
			panic(fmt.Sprintf("%s;缺少expires,且expires大于0", configKey))
		}
	}

	return queueInfo
}
